package net.bwie.flink;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 实时统计订单销售额
 * [bwie@node101 ~]$ nc -lk 9999
 *          u1,o1,milk,6.0
 *          u1,o2,water,5.0
 *          u2,o3,smoke,17.0
 *          u3,o4,bread,5.0
 * @author xuanyu
 * @date 2025/10/27
 */
public class _02FlinkStateDemo {

	public static void main(String[] args) throws Exception{
		// 1.执行环境-env
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1);

		// 2.数据源-source
		DataStreamSource<String> stream = env.socketTextStream("node101", 9999);

		// 3.数据转换-transformation
		// 3-1.map
		SingleOutputStreamOperator<Tuple2<String, Double>> stream31 = stream.map(
			new RichMapFunction<String, Tuple2<String, Double>>() {
				@Override
				public Tuple2<String, Double> map(String value) throws Exception {
					// 数据value -> u1,o1,milk,6.0
					String[] split = value.split(",");
					// 二元组
					return Tuple2.of("销售额", Double.valueOf(split[3]));
				}
			}
		);

		// 3-2.分组
		KeyedStream<Tuple2<String, Double>, String> stream32 = stream31.keyBy(tuple -> tuple.f0);

		// 3-3.聚合
		//SingleOutputStreamOperator<Tuple2<String, Double>> stream33 = stream32.sum(1);
		SingleOutputStreamOperator<Double> stream33 = stream32.map(
			new RichMapFunction<Tuple2<String, Double>, Double>() {
				// todo s1.声明状态
				private transient ValueState<Double> amountState = null ;

				// todo s2.初始化状态
				@Override
				public void open(Configuration parameters) throws Exception {
					amountState = getRuntimeContext().getState(
						new ValueStateDescriptor<Double>("amount-state", Double.class)
					);
				}

				@Override
				public Double map(Tuple2<String, Double> value) throws Exception {
					// 获取当前订单金额
					Double amount = value.f1;

					// todo s3.获取状态值
					Double amountSum = amountState.value();
					// todo s4.使用状态值
					if(null == amountSum){
						amountSum = amount ;
					}else{
						amountSum += amount ;
					}
					// todo s5.更新状态值
					amountState.update(amountSum);

					// 返回最新计算状态值
					return amountSum;
				}
			}
		);

		// 4.数据接收器-sink
		stream33.print();

		// 5.触发执行
		env.execute("FlinkStateDemo") ;
	}

}
